package com.boat.starter.pulsar.core;

import java.net.InetAddress;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.NamedThreadLocal;
import org.springframework.core.env.Environment;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.core.AbstractMessageSendingTemplate;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;

import com.boat.framework.common.monitor.event.PulsarMsgSendFailed;
import com.boat.framework.common.monitor.logs.EventLog;
import com.boat.framework.common.util.JsonUtils;
import com.boat.starter.pulsar.common.PulsarMessage;
import com.boat.starter.pulsar.constant.MQConstant;
import com.boat.starter.pulsar.properties.PulsarProperties;
import com.boat.starter.pulsar.support.PulsarMessageUtils;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;

/**
 * Description:tdmq模板
 *
 * @author: zuomin (myleszelic@outlook.com)
 * @date: 2021/07/16-11:07
 */
@SuppressWarnings({"unchecked", "NullableProblems", "rawtypes"})
public class PulsarTemplate extends AbstractMessageSendingTemplate<String> implements InitializingBean, DisposableBean {

    private static final Logger log = LoggerFactory.getLogger(PulsarTemplate.class);

    private PulsarClient client;
    private PulsarProperties pulsarProperties;
    private Environment environment;
    private MessageConverter messageConverter;

    public PulsarTemplate(PulsarClient client, PulsarProperties pulsarProperties) {
        this.client = client;
        this.pulsarProperties = pulsarProperties;
    }

    private PulsarProducerFactory factory;

    /** 生产者map*/
    private final Map<String, Producer<PulsarMessage>> producerFactories = new ConcurrentHashMap<>(1 << 4);
    
    @Override
    protected void doSend(String destination, Message<?> message) {
        MessageId messageId = sendSync(destination, message);
        if (log.isDebugEnabled()) {
            log.debug("send message to `{}` finished. result:{}", destination, messageId.toString());
        }
    }
    
    /**
     * 设置顺序消息key
     * @param key
     * @return
     */
	public PulsarTemplate key(String key) {
		TemplateLocalVar.keyVar.set(key);
		return this;
	}

	/**
	 * 设置标签
	 * 
	 * 示例代码：
	 * pulsarTemplate.tags("tag1").sendSync(topic, message);
	 * pulsarTemplate.tags("tag1","tag2").sendSync(topic, message);
	 * 
	 * @param name
	 * @return
	 */
	public PulsarTemplate tags(String... tags) {
		if (tags == null) {
			return this;
		}
		
		for(String tag : tags) {
			TemplateLocalVar.propertyVar.get().put(tag, MQConstant.TAGS);
		}
		return this;
	}

	//暂时没使用此方法的场景，为避免增加使用复杂度，先注释
//	/**
//	 * 设置property
//	 * @param name
//	 * @param value
//	 * @return
//	 */
//	public PulsarTemplate property(String name, String value) {
//		TemplateLocalVar.propertyVar.get().put(name, value);
//		return this;
//	}
//	/**
//	 * 设置propertys
//	 * @param properties
//	 * @return
//	 */
//	public PulsarTemplate properties(Map<String, String> properties) {
//		if (properties == null) {
//			return this;
//		}
//		TemplateLocalVar.propertyVar.get().putAll(properties);
//		return this;
//	}
    

    /**
     * <p>Send message in synchronous mode</p>
     *
     * @param topic topic
     * @param message message body
     * @return org.apache.pulsar.client.api.MessageId
     *
     */
    public  MessageId sendSync(String topic, Message message) {
        try {

            Producer<PulsarMessage> producer = getAvailableProducer(topic);
            PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
            //return producer.send(pulsarMessage);
            return newMessage(producer).value(pulsarMessage).send();
        } catch (PulsarClientException e) {
        	//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), e));
			
            log.error("发送同步消息报错,topic:{},错误消息:{}", topic, e.getMessage());
            throw new MessagingException(e.getMessage(), e);
		} catch (Exception ex) {
			//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), ex));
			
			log.error("发送同步消息报错,topic:{},错误消息:{}", topic, ex.getMessage());
			throw new MessagingException(ex.getMessage(), ex);
		}
    }
    
    /**
     * <p>Send message to broker asynchronously</p>
     *
     * @param topic topic
     * @param message message body
     * @return org.apache.pulsar.client.api.MessageId
     */
    public CompletableFuture<MessageId> sendAsync(String topic, Message message) {
        Producer<PulsarMessage> producer = getAvailableProducer(topic);
        PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
        //return producer.sendAsync(this.createPulsarMessage(topic, message));
        return newMessage(producer).value(pulsarMessage).sendAsync();
    }

    /**
     * <p>Send timed message sync</p>
     *
     * @param topic topic
     * @param message message
     * @param deliverAt deliver at
     * @return org.apache.pulsar.client.api.MessageId
     */
    public MessageId sendTimed(String topic, Message message, LocalDateTime deliverAt) {
        try {
            Assert.notNull(deliverAt, "deliverAt can't be null, please set property");
            Producer<PulsarMessage> producer = getAvailableProducer(topic);
            PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
//            return producer.newMessage().value(pulsarMessage)
//                    .deliverAt(deliverAt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
//                    .send();
            return newMessage(producer).value(pulsarMessage)
                    .deliverAt(deliverAt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
                    .send();
        } catch (PulsarClientException e) {
        	//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), e));
			
        	log.error("发送定时消息报错,topic:{},错误消息:{}", topic, e.getMessage());
            throw new MessagingException(e.getMessage(), e);
        } catch (Exception ex) {
        	//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), ex));
			
			log.error("发送定时消息报错,topic:{},错误消息:{}", topic, ex.getMessage());
			throw new MessagingException(ex.getMessage(), ex);
		}
    }

    /**
     * <p>Send timed message async</p>
     * 定时消息的时间范围为当前时间开始计算，864000秒（10天）以内的任意时刻。如10月1日12:00开始，最长可以设置到10月11日12:00。
     *
     * @param topic topic
     * @param message message
     * @param deliverAt deliver at
     * @return org.apache.pulsar.client.api.MessageId
     */
    public CompletableFuture<MessageId> sendTimedAsync(String topic, Message message, LocalDateTime deliverAt) {
        Assert.notNull(deliverAt, "deliverAt can't be null, please set property");
        Producer producer = getAvailableProducer(topic);
        PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
//        return producer.newMessage().value(this.createPulsarMessage(topic, message))
//                .deliverAt(deliverAt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
//                .sendAsync()
//                .exceptionally(e -> {
//                    log.error("send timed message error, topic:{}", topic);
//                    return null;
//                });
        return newMessage(producer).value(pulsarMessage)
                .deliverAt(deliverAt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
                .sendAsync()
                .exceptionally(e -> {
                    log.error("send timed message error, topic:{}", topic);
                    return null;
                });
    }

    /**
     * <p>Send delay message sync</p>
     * 延时消息的时长取值范围为0 - 864000秒（0秒 - 10天）。如10月1日12:00开始，最长可以设置864000秒。如果设置的时间超过这个时间，则直接按864000秒计算，到时会直接投递。
     *
     * @param topic topic
     * @param message message
     * @param deliverAfter deliver after
     * @return org.apache.pulsar.client.api.MessageId
     */
    public MessageId sendDelay(String topic, Message message, Long deliverAfter, TimeUnit timeUnit) {
        try {
            Assert.notNull(deliverAfter, "deliverAfter can't be null, please set property");
            Producer<PulsarMessage> producer = getAvailableProducer(topic);
            PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
//            return producer.newMessage().value(this.createPulsarMessage(topic, message))
//                    .deliverAfter(deliverAfter, timeUnit)
//                    .send();
            return newMessage(producer).value(pulsarMessage)
                    .deliverAfter(deliverAfter, timeUnit)
                    .send();
		} catch (PulsarClientException e) {
			//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), e));
			
			log.error("发送延迟消息报错,topic:{},错误消息:{}", topic, e.getMessage());
			throw new MessagingException(e.getMessage(), e);
		} catch (Exception ex) {
			//记录监控日志
			EventLog.log(new PulsarMsgSendFailed(topic, JsonUtils.toJson(message), ex));
			
			log.error("发送延迟消息报错,topic:{},错误消息:{}", topic, ex.getMessage());
			throw new MessagingException(ex.getMessage(), ex);
		}
    }

    /**
     * <p>Send delay message async</p>
     * 延时消息的时长取值范围为0 - 864000秒（0秒 - 10天）。如10月1日12:00开始，最长可以设置864000秒。如果设置的时间超过这个时间，则直接按864000秒计算，到时会直接投递。
     *
     * @param topic topic
     * @param message message
     * @param deliverAfter deliver after
     * @return org.apache.pulsar.client.api.MessageId
     */
    public CompletableFuture<MessageId> sendDelayAsync(String topic, Message message, Long deliverAfter, TimeUnit timeUnit) {
        Assert.notNull(deliverAfter, "deliverAfter can't be null, please set property");
        Producer producer = getAvailableProducer(topic);
        PulsarMessage pulsarMessage = createPulsarMessage(topic, message);
//        return producer.newMessage().value(this.createPulsarMessage(topic, message))
//                .deliverAfter(deliverAfter, timeUnit)
//                .sendAsync()
//                .exceptionally(e -> {
//                    log.error("send timed message error, topic:{}", topic);
//                    return null;
//                });
        return newMessage(producer).value(pulsarMessage)
                .deliverAfter(deliverAfter, timeUnit)
                .sendAsync()
                .exceptionally(e -> {
                    log.error("send timed message error, topic:{}", topic);
                    return null;
                });
    }

    private Producer<PulsarMessage> getAvailableProducer(String topic) {
        String producerName = factory.getProducerName(topic);
        Producer<PulsarMessage> producer = producerFactories.get(producerName);
        if(Objects.isNull(producer)) {
            producer = factory.create(topic);
            producerFactories.put(producerName, producer);
        }
        return producer;
    }

    /**
     * 创建新消息，初始化key和tag发送方式
     * @param producer
     * @return
     */
    private TypedMessageBuilder<PulsarMessage> newMessage(Producer<PulsarMessage> producer){
    	TypedMessageBuilder<PulsarMessage> newMessage = producer.newMessage();
    	
    	try {
    		//设置key
        	String key = TemplateLocalVar.keyVar.get();
        	if(StrUtil.isNotBlank(key)) {
        		newMessage.key(key);
            }
        	
        	//设置tag或properties
        	Map<String, String> properties = TemplateLocalVar.propertyVar.get();
			if (ObjectUtil.isNotEmpty(properties)) {
				newMessage.properties(properties);
			}
    	}finally {
    		TemplateLocalVar.clear();
		}
    	
    	
    	return newMessage;
    }

	private PulsarMessage createPulsarMessage(String topic, Message message) {
		// 在消息头里面，也加上tags
		Map<String, Object> newHeaders = new HashMap<>();
		newHeaders.putAll(message.getHeaders());
		Map<String, String> properties = TemplateLocalVar.propertyVar.get();
		if (ObjectUtil.isNotEmpty(properties)) {
			newHeaders.put(MQConstant.Header.TAGS, properties.keySet());
		}

		Message<?> msg = this.doConvert(message.getPayload(), newHeaders, null);
		// Do not use getMessageConvert(), because don't want to use
		// getMessageConverter.doConvert()
		return PulsarMessageUtils.convertToPulsarMessage(this.messageConverter, msg);
	}

    @Override
    public void destroy() throws Exception {
        if (!CollectionUtils.isEmpty(producerFactories)) {
            for (Map.Entry<String, Producer<PulsarMessage>> producerEntry : producerFactories.entrySet()) {
                Producer<PulsarMessage> producer = producerEntry.getValue();
                if (Objects.nonNull(producer) && !producer.isConnected()) {
                    producer.close();
                }
            }
            logger.info("Pulsar producer destroy finish");
        }
    }

    @Override
    protected Message<?> doConvert(Object payload, Map<String, Object> headers, MessagePostProcessor postProcessor) {
		// 把内容转换为json字符串
//		String payloadByJson = JSONUtil.toJsonStr(payload);
		String payloadByJson = JsonUtils.toJson(payload);

    	Message<?> message = super.doConvert(payloadByJson, headers, postProcessor);
    	
    	//追加公共头信息
        MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
        //此时已自动创建了头：id和timestamp，timestamp为long型，且不可被重置
        
        //增加其他通用头信息
        //设置更可读的消息创建日期
        builder.setHeader(MQConstant.Header.CREATE_TIME, DateUtil.now());
        builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE);
        builder.setHeaderIfAbsent(MQConstant.Header.PRODUCER_VERSION, MQConstant.VERSION);
        builder.setHeaderIfAbsent(MQConstant.Header.PRODUCER_APP_NAME, environment.getProperty("spring.application.name", "unknown"));
    	String producerIp="";
    	try {
			producerIp = InetAddress.getLocalHost().getHostAddress();
		} catch (Exception ex) {}
    	builder.setHeaderIfAbsent(MQConstant.Header.PRODUCER_IP, producerIp);
    	
        return builder.build();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        factory = new PulsarProducerFactory(client, environment, pulsarProperties);
    }

    public PulsarClient getClient() {
        return client;
    }

    public PulsarProperties getTdmqProperties() {
        return pulsarProperties;
    }

    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    public void initMessageConvert(MessageConverter messageConverter) {
        this.messageConverter = messageConverter;
    }
    
	static class TemplateLocalVar {
		// 顺序消息线程本地变量
		private static final ThreadLocal<String> keyVar = new NamedThreadLocal<>("顺序消息线程本地变量");
		
		// 属性变量线程本地变量
		private static final ThreadLocal<Map<String, String>> propertyVar = ThreadLocal.withInitial(HashMap::new);

		public static void clear() {
			keyVar.remove();
			propertyVar.remove();
		}
	}
}
